Greptile Summary

This PR adds Spark Connect phase 3 metadata reporting to both the profiler and qualification tools, correlating Connect sessions/operations with SQL executions and Spark jobs, persisting lifecycle metadata as per-app CSVs, and optionally writing statement payloads as sidecar .txt files. Previously identified concerns — O(n×m) session-op count scan, local-filesystem-only sidecar writes, missing empty-token filter on spark.job.tags, and load_connect_statement path traversal — are all addressed in this diff.

Confidence Score: 5/5

Safe to merge; no blocking issues found after full review of all changed files.

All previously flagged P1 concerns (Hadoop filesystem bypass, path traversal in Python reader, O(n×m) operation count, missing empty-token filter) are addressed in the current diff. No new P1 or P0 issues were identified. The implementation is consistent with existing patterns, column counts match between Scala headers and YAML catalogs, and the test suite covers correctness, sanitization, and golden roundtrips.

No files require special attention.

Important Files Changed

Sequence Diagram

sequenceDiagram
    participant EL as Event Log
    participant EPB as EventProcessorBase
    participant AB as AppBase
    participant CEH as ConnectEventHandler
    participant P as Profiler/QualRawReportGenerator
    participant CSW as ConnectStatementWriter
    participant FS as Hadoop FileSystem

    EL->>CEH: ConnectOperation/Session events
    CEH->>AB: connectSessions/connectOperations.put(...)
    CEH->>AB: jobTagToConnectOpId.put(jobTag → opId)

    EL->>EPB: SparkListenerSQLExecutionStart
    EPB->>AB: isConnectMode?
    AB-->>EPB: true
    EPB->>EPB: readJobTagsFromSQLStartEvent (reflection, Spark 3.5+)
    EPB->>AB: operationIdToSqlIds.getOrElseUpdate(opId).add(sqlId)

    EL->>EPB: SparkListenerJobStart
    EPB->>AB: isConnectMode?
    AB-->>EPB: true
    EPB->>AB: operationIdToJobIds.getOrElseUpdate(opId).add(jobId)

    P->>P: writeConnectTables(writer, app, writeStatements, hadoopConf)
    P->>P: groupBy sessionId → opCountBySession
    P->>P: build ConnectSessionProfileResult rows
    P->>P: build ConnectOperationProfileResult rows
    alt writeStatementSidecars = true
        P->>CSW: writeStatementFiles(rootDir, ops, hadoopConf)
        CSW->>CSW: sanitize operationId → basename
        CSW->>CSW: require target.getParent == subDirPath
        CSW->>FS: ToolTextFileWriter.write(text)
        CSW-->>P: Map[operationId → basename]
    end
    P->>FS: writeCSVTable(Connect Sessions, sessionRows)
    P->>FS: writeCSVTable(Connect Operations, opRows)

_{Reviews (8): Last reviewed commit: "fix: add Spark Connect runtime dependenc..." | Re-trigger Greptile}

@greptile review

@greptileai review

LGTME mostly. I had a concern about the version of Spark JAR used by Tools. Incase Spark Connect JARs are not present, how do we handle the processing?

LGTME mostly. I had a concern about the version of Spark JAR used by Tools. Incase Spark Connect JARs are not present, how do we handle the processing?

@parthosa Very valid point, thanks for pointing out. Currently tools includes the spark-3.5.7-bin-hadoop3.tgz as default which does not include the connect jars.
Connect distribution has been to be added as a jar dependency separately through maven path.
Updated the user_tools runtime dependency to include the connect jar download step and adding it to classpath.

FYI - even in case of missing jar, the current behavior pops a ClassNotFound error and move onto the next event